use std::any::Any;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use async_trait::async_trait;
use tokio::sync::RwLock;
use tokio::time::sleep;

use protoactor_rs::actor::{Actor, ActorContext, ActorError, DefaultActorContext, Props};
use protoactor_rs::system::ActorSystem;

/// 集群成员状态
#[derive(Debug, Clone, PartialEq)]
enum MemberStatus {
    Alive,
    Unreachable,
    Down,
    Left,
}

/// 集群成员信息
#[derive(Debug, Clone)]
struct Member {
    id: String,
    address: String,
    port: u16,
    status: MemberStatus,
    roles: Vec<String>,
}

impl Member {
    fn new(id: &str, address: &str, port: u16, roles: Vec<String>) -> Self {
        Self {
            id: id.to_string(),
            address: address.to_string(),
            port,
            status: MemberStatus::Alive,
            roles,
        }
    }
}

/// 集群事件基础特性
trait ClusterEvent: Send + Sync + 'static {
    fn clone_boxed(&self) -> Box<dyn ClusterEvent>;
}

/// 成员上线事件
#[derive(Debug, Clone)]
struct MemberJoinedEvent {
    member: Member,
}

impl ClusterEvent for MemberJoinedEvent {
    fn clone_boxed(&self) -> Box<dyn ClusterEvent> {
        Box::new(self.clone())
    }
}

/// 成员变为不可达事件
#[derive(Debug, Clone)]
struct MemberUnreachableEvent {
    member: Member,
}

impl ClusterEvent for MemberUnreachableEvent {
    fn clone_boxed(&self) -> Box<dyn ClusterEvent> {
        Box::new(self.clone())
    }
}

/// 成员变为已下线事件
#[derive(Debug, Clone)]
struct MemberDownEvent {
    member: Member,
}

impl ClusterEvent for MemberDownEvent {
    fn clone_boxed(&self) -> Box<dyn ClusterEvent> {
        Box::new(self.clone())
    }
}

/// 成员离开事件
#[derive(Debug, Clone)]
struct MemberLeftEvent {
    member: Member,
}

impl ClusterEvent for MemberLeftEvent {
    fn clone_boxed(&self) -> Box<dyn ClusterEvent> {
        Box::new(self.clone())
    }
}

/// 集群状态
#[derive(Debug, Clone)]
struct ClusterState {
    members: HashMap<String, Member>,
}

impl ClusterState {
    fn new() -> Self {
        Self {
            members: HashMap::new(),
        }
    }

    fn add_member(&mut self, member: Member) {
        self.members.insert(member.id.clone(), member);
    }

    fn update_member_status(&mut self, id: &str, status: MemberStatus) -> Option<Member> {
        if let Some(member) = self.members.get_mut(id) {
            member.status = status;
            return Some(member.clone());
        }
        None
    }

    fn remove_member(&mut self, id: &str) -> Option<Member> {
        self.members.remove(id)
    }

    fn get_members(&self) -> Vec<Member> {
        self.members.values().cloned().collect()
    }

    fn get_members_by_role(&self, role: &str) -> Vec<Member> {
        self.members
            .values()
            .filter(|m| m.status == MemberStatus::Alive && m.roles.contains(&role.to_string()))
            .cloned()
            .collect()
    }
}

/// 集群Actor接收的消息
#[derive(Debug, Clone)]
enum ClusterActorMessage {
    Join(Member),
    CheckStatus,
    GetMembers,
    GetMembersByRole(String),
    SimulateUnreachable(String),
    SimulateDown(String),
    SimulateLeft(String),
}

/// 集群Actor - 管理集群状态
struct ClusterActor {
    state: Arc<RwLock<ClusterState>>,
    subscribers: Arc<Mutex<Vec<Box<dyn Fn(Box<dyn ClusterEvent>) + Send + Sync>>>>,
}

impl ClusterActor {
    fn new() -> Self {
        Self {
            state: Arc::new(RwLock::new(ClusterState::new())),
            subscribers: Arc::new(Mutex::new(Vec::new())),
        }
    }

    fn publish_event(&self, event: Box<dyn ClusterEvent>) {
        if let Ok(subscribers) = self.subscribers.lock() {
            for subscriber in subscribers.iter() {
                subscriber(event.clone_boxed());
            }
        }
    }
}

impl Clone for ClusterActor {
    fn clone(&self) -> Self {
        Self {
            state: self.state.clone(),
            subscribers: self.subscribers.clone(),
        }
    }
}

#[async_trait]
impl Actor for ClusterActor {
    type Context = DefaultActorContext;

    async fn receive(&self, ctx: &mut Self::Context, msg: Box<dyn Any + Send>) -> Result<(), ActorError> {
        if let Some(join_msg) = msg.downcast_ref::<ClusterActorMessage>() {
            match join_msg {
                ClusterActorMessage::Join(member) => {
                    println!("成员加入: {} at {}:{}", member.id, member.address, member.port);
                    
                    let mut state = self.state.write().await;
                    state.add_member(member.clone());
                    
                    // 发布成员加入事件
                    let event = MemberJoinedEvent {
                        member: member.clone(),
                    };
                    self.publish_event(Box::new(event));
                    
                    // 回复发送者
                    ctx.respond("已加入");
                },
                ClusterActorMessage::CheckStatus => {
                    println!("检查集群状态");
                    
                    let state = self.state.read().await;
                    let members = state.get_members();
                    
                    println!("集群成员 ({}):", members.len());
                    for member in &members {
                        println!("  - {}: {:?} at {}:{}", member.id, member.status, member.address, member.port);
                    }
                    
                    // 回复发送者
                    ctx.respond(format!("集群有 {} 个成员", members.len()));
                },
                ClusterActorMessage::GetMembers => {
                    let state = self.state.read().await;
                    let members = state.get_members();
                    ctx.respond(members);
                },
                ClusterActorMessage::GetMembersByRole(role) => {
                    let state = self.state.read().await;
                    let members = state.get_members_by_role(role);
                    ctx.respond(members);
                },
                ClusterActorMessage::SimulateUnreachable(id) => {
                    println!("模拟成员不可达: {}", id);
                    
                    let mut state = self.state.write().await;
                    if let Some(member) = state.update_member_status(&id, MemberStatus::Unreachable) {
                        // 发布成员不可达事件
                        let event = MemberUnreachableEvent {
                            member: member.clone(),
                        };
                        self.publish_event(Box::new(event));
                        
                        ctx.respond(format!("成员 {} 已标记为不可达", id));
                    } else {
                        ctx.respond(format!("找不到成员 {}", id));
                    }
                },
                ClusterActorMessage::SimulateDown(id) => {
                    println!("模拟成员下线: {}", id);
                    
                    let mut state = self.state.write().await;
                    if let Some(member) = state.update_member_status(&id, MemberStatus::Down) {
                        // 发布成员下线事件
                        let event = MemberDownEvent {
                            member: member.clone(),
                        };
                        self.publish_event(Box::new(event));
                        
                        ctx.respond(format!("成员 {} 已标记为下线", id));
                    } else {
                        ctx.respond(format!("找不到成员 {}", id));
                    }
                },
                ClusterActorMessage::SimulateLeft(id) => {
                    println!("模拟成员离开: {}", id);
                    
                    let mut state = self.state.write().await;
                    if let Some(member) = state.update_member_status(&id, MemberStatus::Left) {
                        // 发布成员离开事件
                        let event = MemberLeftEvent {
                            member: member.clone(),
                        };
                        self.publish_event(Box::new(event));
                        
                        // 移除成员
                        state.remove_member(&id);
                        
                        ctx.respond(format!("成员 {} 已离开集群", id));
                    } else {
                        ctx.respond(format!("找不到成员 {}", id));
                    }
                },
            }
        }
        
        Ok(())
    }
    
    async fn started(&self, _ctx: &mut Self::Context) -> Result<(), ActorError> {
        println!("集群Actor已启动");
        Ok(())
    }
}

/// 集群订阅者Actor，演示如何监听集群事件
#[derive(Clone)]
struct ClusterSubscriberActor {
    name: String,
}

impl ClusterSubscriberActor {
    fn new(name: &str) -> Self {
        Self {
            name: name.to_string(),
        }
    }
}

#[async_trait]
impl Actor for ClusterSubscriberActor {
    type Context = DefaultActorContext;

    async fn receive(&self, _ctx: &mut Self::Context, msg: Box<dyn Any + Send>) -> Result<(), ActorError> {
        if let Some(event) = msg.downcast_ref::<MemberJoinedEvent>() {
            println!("[{}] 收到成员加入事件: {} ({}:{})", 
                self.name, event.member.id, event.member.address, event.member.port);
        } else if let Some(event) = msg.downcast_ref::<MemberUnreachableEvent>() {
            println!("[{}] 收到成员不可达事件: {}", self.name, event.member.id);
        } else if let Some(event) = msg.downcast_ref::<MemberDownEvent>() {
            println!("[{}] 收到成员下线事件: {}", self.name, event.member.id);
        } else if let Some(event) = msg.downcast_ref::<MemberLeftEvent>() {
            println!("[{}] 收到成员离开事件: {}", self.name, event.member.id);
        }
        
        Ok(())
    }
    
    async fn started(&self, _ctx: &mut Self::Context) -> Result<(), ActorError> {
        println!("[{}] 集群订阅者已启动", self.name);
        Ok(())
    }
}

/// 演示如何使用集群
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("集群示例");
    println!("注意：这只是一个模拟演示，实际分布式功能尚未实现");
    
    // 创建系统
    let system = ActorSystem::new()?;
    
    // 创建集群Actor
    let cluster_actor = ClusterActor::new();
    let props = Props::from_producer(move || cluster_actor.clone());
    let cluster_pid = system.spawn::<ClusterActor>(props).await?;
    
    // 创建订阅者Actor
    let subscriber = ClusterSubscriberActor::new("Subscriber1");
    let props = Props::from_producer(move || subscriber.clone());
    let _subscriber_pid = system.spawn::<ClusterSubscriberActor>(props).await?;
    
    // 模拟添加集群成员
    let node1 = Member::new("node1", "127.0.0.1", 8080, vec!["web".to_string()]);
    let join_msg = ClusterActorMessage::Join(node1);
    system.send(&cluster_pid, join_msg).await?;
    
    // 等待一会儿
    sleep(Duration::from_millis(100)).await;
    
    let node2 = Member::new("node2", "127.0.0.1", 8081, vec!["worker".to_string()]);
    let join_msg = ClusterActorMessage::Join(node2);
    system.send(&cluster_pid, join_msg).await?;
    
    sleep(Duration::from_millis(100)).await;
    
    let node3 = Member::new("node3", "127.0.0.1", 8082, vec!["worker".to_string(), "db".to_string()]);
    let join_msg = ClusterActorMessage::Join(node3);
    system.send(&cluster_pid, join_msg).await?;
    
    sleep(Duration::from_millis(100)).await;
    
    // 检查状态
    let check_msg = ClusterActorMessage::CheckStatus;
    system.send(&cluster_pid, check_msg).await?;
    
    sleep(Duration::from_secs(1)).await;
    
    // 模拟节点不可达
    let unreachable_msg = ClusterActorMessage::SimulateUnreachable("node2".to_string());
    system.send(&cluster_pid, unreachable_msg).await?;
    
    sleep(Duration::from_millis(500)).await;
    
    // 模拟节点下线
    let down_msg = ClusterActorMessage::SimulateDown("node2".to_string());
    system.send(&cluster_pid, down_msg).await?;
    
    sleep(Duration::from_millis(500)).await;
    
    // 模拟节点离开
    let left_msg = ClusterActorMessage::SimulateLeft("node3".to_string());
    system.send(&cluster_pid, left_msg).await?;
    
    sleep(Duration::from_millis(500)).await;
    
    // 再次检查状态
    let check_msg = ClusterActorMessage::CheckStatus;
    system.send(&cluster_pid, check_msg).await?;
    
    // 等待系统处理完所有消息
    sleep(Duration::from_secs(2)).await;
    
    // 停止集群Actor
    system.stop(&cluster_pid).await?;
    
    println!("示例完成");
    
    Ok(())
} 